package org.cliff.message.platform.core.resend;

import cn.hutool.core.collection.CollectionUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import org.cliff.message.platform.core.domain.ReceivedMessageResendDetail;
import org.cliff.message.platform.core.domain.SendStatusConstants;
import org.cliff.message.platform.core.mapper.ReceivedMessageResendDetailMapper;
import org.cliff.message.platform.core.service.ReceivedMessageResendService;
import org.cliff.message.platform.core.service.ReceivedMessageService;
import org.cliff.message.platform.core.util.IDistributeLock;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestTemplate;

import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 接收到消息重发器
 * 
 * @author 刘文
 * @date 2023/10/25
 */
@Component
@Slf4j
public class ReceivedMessageResend {

    @Autowired
    private ReceivedMessageResendDetailMapper resendDetailMapper;

    @Autowired
    private IDistributeLock distributeLock;

    @Autowired
    private ReceivedMessageResendService resendService;

    @Autowired
    private ReceivedMessageService receivedMessageService;

    @Autowired
    private RestTemplate mpResendRestTemplate;

    /**
     * 重发收到消息分布式锁
     */
    private static final String RESEND_RECEIVED_MESSAGE_DISTRIBUTE_LOCK_NAME = "msg_plat:resendReceivedMessage";

    /**
     * 定时任务每两分钟扫描需要重发的接收到/消费的消息
     */
    @Scheduled(cron = "0 0/2 * * * ?")
    public void resendMessage() {
        long startTime = System.currentTimeMillis();
        // 持有100s，小于2分钟的间隔
        boolean locked =
            distributeLock.tryLock(RESEND_RECEIVED_MESSAGE_DISTRIBUTE_LOCK_NAME, 500, 100000, TimeUnit.MILLISECONDS);
        if (!locked) {
            log.debug("Return for didn't get distributed lock");
            return;
        }

        try {
            LambdaQueryWrapper<ReceivedMessageResendDetail> queryWrapper =
                Wrappers.lambdaQuery(ReceivedMessageResendDetail.class)
                    .eq(ReceivedMessageResendDetail::getSendStatus, SendStatusConstants.PENDING)
                    .le(ReceivedMessageResendDetail::getPlanedSendTime, System.currentTimeMillis());
            List<ReceivedMessageResendDetail> resendMessages = resendDetailMapper.selectList(queryWrapper);
            if (CollectionUtil.isEmpty(resendMessages)) {
                log.debug("No message to resend");
                // 等待1秒钟，避免其他节点抢占锁重复查询。
                Thread.sleep(1000);
                return;
            }

            log.info("Have {} received messages to resend.", resendMessages.size());

            List<ReceivedMessageResendTask> resendTasks =
                resendMessages.stream().map(resendMessage -> new ReceivedMessageResendTask(resendMessage, resendService,
                    receivedMessageService, mpResendRestTemplate)).collect(Collectors.toList());
            int submittedTasks = ResendMessageExecutor.RECEIVED_MESSAGE_RESEND_EXECUTOR
                    .executeTasks(resendTasks, startTime + 90000);
            log.info("Finished one batch message resend, retrieved {} tasks and submitted {} tasks.",
                    resendMessages.size(), submittedTasks);
        } catch (InterruptedException e) {
            log.error("Interrupted while sleeping.", e);
        } finally {
            distributeLock.unlock(RESEND_RECEIVED_MESSAGE_DISTRIBUTE_LOCK_NAME);
        }
    }
}
